package com.kingsoft.dc.khaos.module.mpp.operator;

import com.kingsoft.dc.khaos.module.mpp.client.ConfVars;
import com.kingsoft.dc.khaos.module.mpp.client.KhaosContants;
import com.kingsoft.dc.khaos.module.mpp.client.MppClient;
import com.kingsoft.dc.khaos.module.mpp.exception.MppSqlException;
import com.kingsoft.dc.khaos.module.mpp.metadata.CombineStrategy;
import com.kingsoft.dc.khaos.module.mpp.metadata.MppConfig;
import com.kingsoft.dc.khaos.module.mpp.model.ColumnEntity;
import com.kingsoft.dc.khaos.module.mpp.sqlbuilder.SqlBuilderUtils;
import com.kingsoft.dc.khaos.plugin.core.KhaosContext;
import org.apache.commons.lang3.StringUtils;

import java.util.Arrays;
import java.util.List;

/**
 * 当前表.
 */
public class TyCurrentTab extends CombineStrategy {
    private String rightTable = "";

    @Override
    public void execute(KhaosContext kc, String moduleId, MppConfig mppConfig) {
        //初始化MPP client,client已经初始化数据库连接
        MppClient client = null;
        try {
            client = new MppClient(mppConfig, kc);

            //获取操作的库表参数
            //比如获取源表所属的schema
            String srcSchema = String.format("\"%s\"", mppConfig.getSrcSchema());
            log().info("源库名srcSchema===>  " + srcSchema);
            //源表名
            String srcTable = String.format("\"%s\"", mppConfig.getSrcTable());
            log().info("源表名srcTable===>  " + srcTable);
            //源表过滤条件
            String filter = mppConfig.getDataFilter();

            //目标表所属schema名
            String tarSchema = String.format("\"%s\"", mppConfig.getTarSchema());
            log().info("目标库名tarSchema===>  " + tarSchema);
            //目标表名
            String tarTable = String.format("\"%s\"", mppConfig.getTarTable());
            log().info("目标表名tarTable===>  " + tarTable);

            //减档表所属schema名
            String delSchema = String.format("\"%s\"", mppConfig.getDelSchema());
            log().info("减档库名delSchema===>  " + delSchema);
            //减档表名
            String delTable = String.format("\"%s\"", mppConfig.getDelTable());
            log().info("减档表名delTable===>  " + delTable);

            //拼接加1的右表
            rightTable = String.format("\"%s1\"", mppConfig.getTarTable());

            //主键列表
            List<String> keys = mppConfig.getpKeys();

            //获取源档类型
            String src_Type = mppConfig.getSrcType();

            //获取主键字符串
            String pKeys = SqlBuilderUtils.selKeys(tarTable, keys);
            log().info("目标表主键字符串===>  " + pKeys);

            //是否物理删除
            boolean isDeleted = mppConfig.isDeleted();

            //获取追加字段名 比如获取[数据日期]字段名
            String bizDateColName = client.getAppendColName(ConfVars.P9_DATA_DATE);

            //获取追加字段值，比如获取[数据日期]字段值
            String bizDate = kc.getConf().getString(KhaosContants.BIZ_DATE);
            String batchNo = kc.getConf().getString(KhaosContants.BATCH_NO);
            String jobName = kc.getConf().getString(KhaosContants.JOB_NAME);
            String jobId = kc.getConf().getString(KhaosContants.JOB_INSTANCE_ID);
            //获取表全字段信息如id,name
            List<String> columnEntities = client.getColumnNameList(mppConfig.getSrcSchema(), mppConfig.getSrcTable());
            log().info("源表全字段信息===>  " + StringUtils.join(columnEntities, ","));

            //获取表及表全字段信息
            List<ColumnEntity> tableColumns = client.getColumnInfo(mppConfig.getSrcSchema(), mppConfig.getSrcTable());
            //获取目标表主键字段信息
            String columns = SqlBuilderUtils.selKeys(tarTable, columnEntities);
            log().info("目标表主键字段信息===>  " + columns);
            //获取两表关联主键SQL字符串
            String pKs = SqlBuilderUtils.joinKeys(srcTable, tarTable, keys);
            //获取两表关联字段空值处理SQL字符串
            String columnNulls = SqlBuilderUtils.replaceNull(srcTable, tarTable, tableColumns);

            //获取两表关联字段空值处理SQL字符串
            String columnNst = SqlBuilderUtils.replaceNull(srcTable, rightTable, tableColumns);
            log().info("两表关联字段空值处理SQL字符串===>  " + columnNst);
            //获取源表别名及源表全字段信息如s.id,s.name
            String selTabCols = SqlBuilderUtils.selTabCols(srcTable, columnEntities);
            log().info("源表别名及源表全字段信息===>  " + selTabCols);
            //获取目标表别名及目标表全字段信息如s.id,s.name
            String selTabColst = SqlBuilderUtils.selTabCols(tarTable, columnEntities);
            log().info("目标表别名及目标表全字段信息===>  " + selTabColst);
            //获取表别名+表全字段信息如s.id,s.name
            // String selTabColst1 = SqlBuilderUtils.selTabCols(rightTable,columnEntities);

            /**拼接当前全量表减档特殊算子SQL
             String exeu_sql = buildSqlDel(src_Type,tarSchema,tarTable,delSchema,delTable,bizDate,jobName,batchNo,pKeys,isDeleted, client);
             System.out.println(exeu_sql);
             client.execInsert(exeu_sql);
             **/
            String exe_sql = "";
            if (null == filter || filter.equals("")) {
                //拼接当前全量表算子SQL
                exe_sql = buildSql(src_Type, pKeys, tarSchema, tarTable, srcSchema, srcTable, delSchema, delTable, bizDate, jobName, batchNo, jobId, columns, pKs, columnNst, keys, selTabCols, selTabColst, isDeleted, client);
            } else {
                exe_sql = buildSqlWithFilter(src_Type, pKeys, tarSchema, tarTable, srcSchema, srcTable, delSchema, delTable, bizDate, jobName, batchNo, jobId, columns, pKs, columnNst, keys, selTabCols, selTabColst, isDeleted, client, filter);
            }

            client.execInsert(exe_sql);
        } catch (Exception e) {
            e.printStackTrace();
            throw new MppSqlException("执行MPP SQL 异常");
        } finally {
            if (client != null) {
                client.close();
            }
        }
    }

    /**
     * 拼接当前全量表算子SQL实现类
     * public String buildSqlDel(String src_Type,String tarSchema,String tarTable,String delSchema,String delTable,String bizDate,String jobName,String batchNo,String pKeys,boolean isDeleted, MppClient client) {
     * String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
     * String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
     * String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
     * String tyDelDate = client.getAppendColName(ConfVars.P9_DEL_DATE);
     * String tyDelBatch = client.getAppendColName(ConfVars.P9_DEL_BATCH);
     * String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
     * String del_sql = "";
     * //源档是增量配置物理删除
     * if (src_Type.equalsIgnoreCase("ADD") && isDeleted == true){
     * del_sql = String.format("update %s.%s \n" +
     * "\t   set " + tyDataDate + " = to_date('%s','YYYYMMDD'),\n" +
     * "\t       " + tyJobName + " = '%s',\n" +
     * "\t       " + tyBatchNumber + " = '%s',\n" +
     * "\t       " + tyDelFlag + " = '1',\n" +
     * "\t       " + tyDelDate + " = to_date('%s','YYYYMMDD'),\n" +
     * "\t       " + tyDelBatch + " = '%s'\n" +
     * "\t   WHERE (%s) in (select %s from %s.%s); --#PKS#表示所有主键\n",tarSchema, tarTable,bizDate,jobName,batchNo,bizDate,batchNo,pKeys,pKeys,delSchema,delTable);
     * }
     * return del_sql;
     * }
     **/

    //拼接当前全量表算子SQL实现类
    public String buildSql(String src_Type,
                           String pKeys,
                           String tarSchema,
                           String tarTable,
                           String srcSchema,
                           String srcTable,
                           String delSchema,
                           String delTable,
                           String bizDate,
                           String jobName,
                           String batchNo,
                           String jobId,
                           String columns,
                           String pKs,
                           String columnNst,
                           List<String> keys,
                           String selTabCols,
                           String selTabColst,
                           boolean isDeleted,
                           MppClient client) {
        System.out.println("TyCurrentTab buildSql start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyDelDate = client.getAppendColName(ConfVars.P9_DEL_DATE);
        String tyDelBatch = client.getAppendColName(ConfVars.P9_DEL_BATCH);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String tyRecInitDate = client.getAppendColName(ConfVars.P9_REC_INIT_DATE);


        String all_sql = "";
        //创建动态临时表
        String createvt_sql = String.format("drop table if exists %s.tmp_%s; \n", tarSchema, jobId)
                + String.format("create table %s.tmp_%s as select * from  %s.%s limit 0; \n", tarSchema, jobId, tarSchema, tarTable)
                + String.format("truncate table %s.tmp_%s; \n", tarSchema, jobId);
        //删除动态临时表
        String drop_tmptable_sql = String.format("drop table if exists %s.tmp_%s; \n", tarSchema, jobId);
        //全量档
        if (src_Type.equalsIgnoreCase("ALL")) {
            //插入不变或者更新的记录至临时表
            String exe_sql_one = String.format("insert into %s.tmp_%s (\n" +
                            "\t %s,\n" +
                            "\t " + tyDataDate + ",\n" +
                            "\t " + tyJobName + ",\t\n" +
                            "\t " + tyBatchNumber + ",\n" +
                            "\t " + tyDelFlag + ",\n" +
                            "\t " + tyDelDate + ",\n" +
                            "\t " + tyDelBatch + ",\n" +
                            "\t " + tyRecInitDate + "\n" +
                            "\t ) select \n" +
                            "\t      %s,\n" +
                            "\t\t    case when %s.%s is not null \n" +
                            "\t\t         then %s." + tyDataDate + " \n" +
                            "\t\t         else to_date('%s','YYYYMMDD') end,\n" +
                            "\t\t    case when %s.%s is not null \n" +
                            "\t\t         then %s." + tyJobName + " \n" +
                            "\t\t         else '%s' end,\n" +
                            "\t\t    case when %s.%s is not null \n" +
                            "\t\t         then %s." + tyBatchNumber + " \n" +
                            "\t\t         else '%s' end,\n" +
                            "\t\t    '0' AS " + tyDelFlag + ",\n" +
                            "\t\t    to_date('00010101','YYYYMMDD') AS " + tyDelDate + ",\n" +
                            "\t\t    '0000' AS " + tyDelBatch + ",\n" +
                            "\t\t    %s." + tyRecInitDate + "\n" +
                            "\t from %s.%s %s\n" +
                            "\t inner join %s.%s %s \n" +
                            "\t on %s \n" +
                            "\t left join %s.%s %s\n" +
                            "\t on %s ;\n", tarSchema, jobId, columns, selTabCols, rightTable, keys.get(0), rightTable, bizDate, rightTable, keys.get(0)
                    , rightTable, jobName, rightTable, keys.get(0), rightTable, batchNo, tarTable, srcSchema, srcTable, srcTable, tarSchema, tarTable, tarTable, pKs, tarSchema, tarTable, rightTable, columnNst);
            //插入新增记录至临时表
            String exe_sql_two = String.format("insert into %s.tmp_%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                    "\t      '%s' as " + tyJobName + ",\t\n" +
                    "\t      '%s' as " + tyBatchNumber + ",\n" +
                    "\t      '0' AS " + tyDelFlag + ",\n" +
                    "\t      to_date('00010101','YYYYMMDD') as " + tyDelDate + ",\n" +
                    "\t      '0000' AS " + tyDelBatch + ",\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyRecInitDate + "\n" +
                    "\t\tFROM %s.%s %s\n" +
                    "\t\tLEFT JOIN %s.%s %s\n" +
                    "\t\ton %s\n" +
                    "\t\twhere %s.%s is null;\n", tarSchema, jobId, columns, selTabCols, bizDate, jobName, batchNo, bizDate, srcSchema, srcTable, srcTable, tarSchema, tarTable, tarTable, pKs, tarTable, keys.get(0));
            //插入删除的记录（本次删除+历史删除）至临时表
            String exe_sql_three = String.format("insert into %s.tmp_%s (\n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyJobName + ",\t\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tyDelFlag + ",\n" +
                            "\t      " + tyDelDate + ",\n" +
                            "\t      " + tyDelBatch + ",\n" +
                            "\t      " + tyRecInitDate + "\n" +
                            "\t) select \n" +
                            "\t      %s,\n" +
                            "\t      case when %s." + tyDelFlag + " = '1' \n" +
                            "\t\t         then %s." + tyDataDate + " \n" +
                            "\t\t         else to_date('%s','YYYYMMDD') end,\n" +
                            "\t\t    %s." + tyJobName + ",\t\n" +
                            "\t      %s." + tyBatchNumber + ",\n" +
                            "\t      '1' as " + tyDelFlag + ",\n" +
                            "\t      case when %s." + tyDelFlag + " = '1' \n" +
                            "\t           then %s." + tyDelDate + " \n" +
                            "\t           else to_date('%s','YYYYMMDD') end,\n" +
                            "\t      case when %s." + tyDelFlag + " = '1' \n" +
                            "\t           then %s." + tyDelBatch + " \n" +
                            "\t           else '%s' end,\n" +
                            "\t      %s." + tyRecInitDate + "\n" +
                            "\t\tFROM %s.%s %s\n" +
                            "\t\tLEFT JOIN %s.%s %s\n" +
                            "\t\ton %s\n" +
                            "\t\twhere %s.%s is null;\n", tarSchema, jobId, columns, selTabColst, tarTable, tarTable, bizDate, tarTable, tarTable, tarTable, tarTable, bizDate
                    , tarTable, tarTable, batchNo, tarTable, tarSchema, tarTable, tarTable, srcSchema, srcTable, srcTable, pKs, srcTable, keys.get(0));
            //清空目标表
            String exe_sql_four = String.format("delete from %s.%s ;\n", tarSchema, tarTable);
            //将临时表的数据插入到目标表
            String exe_sql_five = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t\tfrom %s.tmp_%s;\n", tarSchema, tarTable, columns, columns, tarSchema, jobId);
            //清空目标表
            String exe_sql_six = String.format("truncate table %s.tmp_%s ;\n", tarSchema, jobId);

            all_sql = "\n BEGIN WORK; \n " + createvt_sql + exe_sql_one + exe_sql_two + exe_sql_three + exe_sql_four + exe_sql_five + "\n COMMIT ; \n " + exe_sql_six + drop_tmptable_sql + "\n COMMIT ; \n ";
        }
        //1.源档是增量或2.源档是增量且有物理删除，因为要处理两个步骤：增，减源表处理
        else if ((src_Type.equalsIgnoreCase("ADD")) || (src_Type.equalsIgnoreCase("ADD") && isDeleted == true)) {
            //插入无变化的数据
            String del_sql = "";
            String exe_sql_a_d_one = String.format("insert into %s.tmp_%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t\tFROM %s.%s %s\n" +
                    "\t\tLEFT JOIN %s.%s %s\n" +
                    "\t\ton %s\n" +
                    "\t\twhere %s.%s is null;\n", tarSchema, jobId, columns, selTabColst, tarSchema, tarTable, tarTable, srcSchema, srcTable, srcTable, pKs, srcTable, keys.get(0));

            //插入更新+新增记录
            String exe_sql_a_d_two = String.format("insert into %s.tmp_%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                    "\t      '%s' as " + tyJobName + ",\t\n" +
                    "\t      '%s' as " + tyBatchNumber + ",\n" +
                    "\t      '0' AS " + tyDelFlag + ",\n" +
                    "\t      to_date('00010101','YYYYMMDD') as " + tyDelDate + ",\n" +
                    "\t      '0000' AS " + tyDelBatch + ",\n" +
                    "\t\t    case when %s.%s is not null \n" +
                    "\t\t         then %s." + tyRecInitDate + " \n" +
                    "\t\t         else to_date('%s','YYYYMMDD') end \n" +
                    "\t\tFROM %s.%s %s\n" +
                    "\t\tLEFT JOIN %s.%s %s\n" +
                    "\t\ton %s ;\n", tarSchema, jobId, columns, selTabCols, bizDate, jobName, batchNo, tarTable, keys.get(0), tarTable, bizDate, srcSchema, srcTable, srcTable, tarSchema, tarTable, tarTable, pKs);

            //源档是增量配置物理删除
            if (src_Type.equalsIgnoreCase("ADD") && isDeleted == true) {
                del_sql = String.format("\n update %s.%s \n" +
                        "\t   set " + tyDataDate + " = to_date('%s','YYYYMMDD'),\n" +
                        "\t       " + tyJobName + " = '%s',\n" +
                        "\t       " + tyBatchNumber + " = '%s',\n" +
                        "\t       " + tyDelFlag + " = '1',\n" +
                        "\t       " + tyDelDate + " = to_date('%s','YYYYMMDD'),\n" +
                        "\t       " + tyDelBatch + " = '%s'\n" +
                        "\t   WHERE (%s) in (select %s from %s.%s); \n", tarSchema, tarTable, bizDate, jobName, batchNo, bizDate, batchNo, pKeys, pKeys, delSchema, delTable);
            }

            //清空目标表
            String exe_sql_a_d_three = String.format("\n delete from %s.%s;\n", tarSchema, tarTable);
            //插入目标表
            String exe_sql_a_d_four = String.format("\n insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t\tfrom %s.tmp_%s;\n", tarSchema, tarTable, columns, columns, tarSchema, jobId);
            //清空目标表 
            String exe_sql_a_d_five = String.format("truncate table %s.tmp_%s ;\n", tarSchema, jobId);

            all_sql = "\n BEGIN WORK; \n " + del_sql + createvt_sql + exe_sql_a_d_one + exe_sql_a_d_two + exe_sql_a_d_three + exe_sql_a_d_four + "\n COMMIT ; \n " + exe_sql_a_d_five + drop_tmptable_sql + "\n COMMIT ; \n ";
        }
        System.out.println("TyCurrentTab buildSql end...");
        return all_sql;
    }

    //拼接当前全量表算子SQL实现类
    public String buildSqlWithFilter(String src_Type,
                                     String pKeys,
                                     String tarSchema,
                                     String tarTable,
                                     String srcSchema,
                                     String srcTable,
                                     String delSchema,
                                     String delTable,
                                     String bizDate,
                                     String jobName,
                                     String batchNo,
                                     String jobId,
                                     String columns,
                                     String pKs,
                                     String columnNst,
                                     List<String> keys,
                                     String selTabCols,
                                     String selTabColst,
                                     boolean isDeleted,
                                     MppClient client,
                                     String filter) {
        System.out.println("TyCurrentTab buildSqlWithFilter start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyDelDate = client.getAppendColName(ConfVars.P9_DEL_DATE);
        String tyDelBatch = client.getAppendColName(ConfVars.P9_DEL_BATCH);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String tyRecInitDate = client.getAppendColName(ConfVars.P9_REC_INIT_DATE);


        String schemaTblFilter = String.format("(select * from %s.%s where %s)", srcSchema, srcTable, filter);
        String all_sql = "";
        //创建动态临时表
        String createvt_sql = String.format("drop table if exists %s.tmp_%s; \n", tarSchema, jobId)
                + String.format("create table %s.tmp_%s as select * from  %s.%s limit 0; \n", tarSchema, jobId, tarSchema, tarTable)
                + String.format("truncate table %s.tmp_%s; \n", tarSchema, jobId);
        //删除动态临时表
        String drop_tmptable_sql = String.format("drop table if exists %s.tmp_%s; \n", tarSchema, jobId);
        //全量档
        if (src_Type.equalsIgnoreCase("ALL")) {
            //插入不变或者更新的记录至临时表
            String exe_sql_one = String.format("insert into %s.tmp_%s (\n" +
                            "\t %s,\n" +
                            "\t " + tyDataDate + ",\n" +
                            "\t " + tyJobName + ",\t\n" +
                            "\t " + tyBatchNumber + ",\n" +
                            "\t " + tyDelFlag + ",\n" +
                            "\t " + tyDelDate + ",\n" +
                            "\t " + tyDelBatch + ",\n" +
                            "\t " + tyRecInitDate + "\n" +
                            "\t ) select \n" +
                            "\t      %s,\n" +
                            "\t\t    case when %s.%s is not null \n" +
                            "\t\t         then %s." + tyDataDate + " \n" +
                            "\t\t         else to_date('%s','YYYYMMDD') end,\n" +
                            "\t\t    case when %s.%s is not null \n" +
                            "\t\t         then %s." + tyJobName + " \n" +
                            "\t\t         else '%s' end,\n" +
                            "\t\t    case when %s.%s is not null \n" +
                            "\t\t         then %s." + tyBatchNumber + " \n" +
                            "\t\t         else '%s' end,\n" +
                            "\t\t    '0' AS " + tyDelFlag + ",\n" +
                            "\t\t    to_date('00010101','YYYYMMDD') AS " + tyDelDate + ",\n" +
                            "\t\t    '0000' AS " + tyDelBatch + ",\n" +
                            "\t\t    %s." + tyRecInitDate + "\n" +
                            "\t from %s %s\n" +
                            "\t inner join %s.%s %s \n" +
                            "\t on %s \n" +
                            "\t left join %s.%s %s\n" +
                            "\t on %s ;\n", tarSchema, jobId, columns, selTabCols, rightTable,
                    keys.get(0), rightTable, bizDate, rightTable, keys.get(0)
                    , rightTable, jobName, rightTable, keys.get(0),
                    rightTable, batchNo, tarTable,
                    schemaTblFilter, srcTable, tarSchema, tarTable, tarTable, pKs, tarSchema, tarTable, rightTable, columnNst);
            //插入新增记录至临时表
            String exe_sql_two = String.format("insert into %s.tmp_%s (\n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyJobName + ",\t\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tyDelFlag + ",\n" +
                            "\t      " + tyDelDate + ",\n" +
                            "\t      " + tyDelBatch + ",\n" +
                            "\t      " + tyRecInitDate + "\n" +
                            "\t) select \n" +
                            "\t      %s,\n" +
                            "\t      to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                            "\t      '%s' as " + tyJobName + ",\t\n" +
                            "\t      '%s' as " + tyBatchNumber + ",\n" +
                            "\t      '0' AS " + tyDelFlag + ",\n" +
                            "\t      to_date('00010101','YYYYMMDD') as " + tyDelDate + ",\n" +
                            "\t      '0000' AS " + tyDelBatch + ",\n" +
                            "\t      to_date('%s','YYYYMMDD') as " + tyRecInitDate + "\n" +
                            "\t\tFROM %s %s\n" +
                            "\t\tLEFT JOIN %s.%s %s\n" +
                            "\t\ton %s\n" +
                            "\t\twhere %s.%s is null;\n", tarSchema, jobId, columns,
                    selTabCols, bizDate, jobName, batchNo, bizDate, schemaTblFilter,
                    srcTable, tarSchema, tarTable, tarTable, pKs, tarTable, keys.get(0));
            //插入删除的记录（本次删除+历史删除）至临时表
            String exe_sql_three = String.format("insert into %s.tmp_%s (\n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyJobName + ",\t\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tyDelFlag + ",\n" +
                            "\t      " + tyDelDate + ",\n" +
                            "\t      " + tyDelBatch + ",\n" +
                            "\t      " + tyRecInitDate + "\n" +
                            "\t) select \n" +
                            "\t      %s,\n" +
                            "\t      case when %s." + tyDelFlag + " = '1' \n" +
                            "\t\t         then %s." + tyDataDate + " \n" +
                            "\t\t         else to_date('%s','YYYYMMDD') end,\n" +
                            "\t\t    %s." + tyJobName + ",\t\n" +
                            "\t      %s." + tyBatchNumber + ",\n" +
                            "\t      '1' as " + tyDelFlag + ",\n" +
                            "\t      case when %s." + tyDelFlag + " = '1' \n" +
                            "\t           then %s." + tyDelDate + " \n" +
                            "\t           else to_date('%s','YYYYMMDD') end,\n" +
                            "\t      case when %s." + tyDelFlag + " = '1' \n" +
                            "\t           then %s." + tyDelBatch + " \n" +
                            "\t           else '%s' end,\n" +
                            "\t      %s." + tyRecInitDate + "\n" +
                            "\t\tFROM %s.%s %s\n" +
                            "\t\tLEFT JOIN %s %s\n" +
                            "\t\ton %s\n" +
                            "\t\twhere %s.%s is null;\n", tarSchema, jobId, columns, selTabColst,
                    tarTable, tarTable, bizDate, tarTable, tarTable, tarTable, tarTable, bizDate
                    , tarTable, tarTable, batchNo, tarTable, tarSchema, tarTable, tarTable,
                    schemaTblFilter, srcTable, pKs, srcTable, keys.get(0));
            //清空目标表
            String exe_sql_four = String.format("delete from %s.%s ;\n", tarSchema, tarTable);
            //将临时表的数据插入到目标表
            String exe_sql_five = String.format("insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t\tfrom %s.tmp_%s;\n", tarSchema, tarTable, columns, columns, tarSchema, jobId);
            //清空目标表
            String exe_sql_six = String.format("truncate table %s.tmp_%s ;\n", tarSchema, jobId);

            all_sql = "\n BEGIN WORK; \n " + createvt_sql + exe_sql_one + exe_sql_two + exe_sql_three + exe_sql_four + exe_sql_five + "\n COMMIT ; \n " + exe_sql_six + drop_tmptable_sql + "\n COMMIT ; \n ";
        }
        //1.源档是增量或2.源档是增量且有物理删除，因为要处理两个步骤：增，减源表处理
        else if ((src_Type.equalsIgnoreCase("ADD")) || (src_Type.equalsIgnoreCase("ADD") && isDeleted == true)) {
            //插入无变化的数据
            String del_sql = "";
            String exe_sql_a_d_one = String.format("insert into %s.tmp_%s (\n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyJobName + ",\t\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tyDelFlag + ",\n" +
                            "\t      " + tyDelDate + ",\n" +
                            "\t      " + tyDelBatch + ",\n" +
                            "\t      " + tyRecInitDate + "\n" +
                            "\t) select \n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyJobName + ",\t\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tyDelFlag + ",\n" +
                            "\t      " + tyDelDate + ",\n" +
                            "\t      " + tyDelBatch + ",\n" +
                            "\t      " + tyRecInitDate + "\n" +
                            "\t\tFROM %s.%s\n" +
                            "\t\tLEFT JOIN %s %s\n" +
                            "\t\ton %s\n" +
                            "\t\twhere %s.%s is null;\n", tarSchema, jobId, columns, selTabColst,
                    tarSchema, tarTable, schemaTblFilter, srcTable, pKs, srcTable, keys.get(0));

            //插入更新+新增记录
            String exe_sql_a_d_two = String.format("insert into %s.tmp_%s (\n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyJobName + ",\t\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tyDelFlag + ",\n" +
                            "\t      " + tyDelDate + ",\n" +
                            "\t      " + tyDelBatch + ",\n" +
                            "\t      " + tyRecInitDate + "\n" +
                            "\t) select \n" +
                            "\t      %s,\n" +
                            "\t      to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                            "\t      '%s' as " + tyJobName + ",\t\n" +
                            "\t      '%s' as " + tyBatchNumber + ",\n" +
                            "\t      '0' AS " + tyDelFlag + ",\n" +
                            "\t      to_date('00010101','YYYYMMDD') as " + tyDelDate + ",\n" +
                            "\t      '0000' AS " + tyDelBatch + ",\n" +
                            "\t\t    case when %s.%s is not null \n" +
                            "\t\t         then %s." + tyRecInitDate + " \n" +
                            "\t\t         else to_date('%s','YYYYMMDD') end \n" +
                            "\t\tFROM %s %s\n" +
                            "\t\tLEFT JOIN %s.%s %s\n" +
                            "\t\ton %s ;\n", tarSchema, jobId, columns, selTabCols, bizDate,
                    jobName, batchNo, tarTable, keys.get(0), tarTable, bizDate,
                    schemaTblFilter, srcTable, tarSchema, tarTable, tarTable, pKs);

            //源档是增量配置物理删除
            if (src_Type.equalsIgnoreCase("ADD") && isDeleted == true) {
                del_sql = String.format("\n update %s.%s \n" +
                                "\t   set " + tyDataDate + " = to_date('%s','YYYYMMDD'),\n" +
                                "\t       " + tyJobName + " = '%s',\n" +
                                "\t       " + tyBatchNumber + " = '%s',\n" +
                                "\t       " + tyDelFlag + " = '1',\n" +
                                "\t       " + tyDelDate + " = to_date('%s','YYYYMMDD'),\n" +
                                "\t       " + tyDelBatch + " = '%s'\n" +
                                "\t   WHERE (%s) in (select %s from %s.%s); \n", tarSchema, tarTable,
                        bizDate, jobName, batchNo, bizDate, batchNo, pKeys, pKeys, delSchema, delTable);
            }

            //清空目标表
            String exe_sql_a_d_three = String.format("\n delete from %s.%s;\n", tarSchema, tarTable);
            //插入目标表
            String exe_sql_a_d_four = String.format("\n insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyJobName + ",\t\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyRecInitDate + "\n" +
                    "\t\tfrom %s.tmp_%s;\n", tarSchema, tarTable, columns, columns, tarSchema, jobId);
            //清空目标表
            String exe_sql_a_d_five = String.format("truncate table %s.tmp_%s ;\n", tarSchema, jobId);

            all_sql = "\n BEGIN WORK; \n " + del_sql + createvt_sql + exe_sql_a_d_one + exe_sql_a_d_two + exe_sql_a_d_three + exe_sql_a_d_four + "\n COMMIT ; \n " + exe_sql_a_d_five + drop_tmptable_sql + "\n COMMIT ; \n ";
        }
        System.out.println("TyCurrentTab buildSqlWithFilter end...");
        return all_sql;
    }

}
